package com.shujia.transformation

import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._

object Demo04KeyBy {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 读取students.txt 过滤出文科班的学生
    val stuDS: DataStream[String] = env.readTextFile("Flink/data/stu/students.txt")

    // 统计每个班级的性别人数
    stuDS
      .map(line => {
        val stuArr: Array[String] = line.split(",")
        (stuArr(4) + "," + stuArr(3), 1)
      })
      .keyBy(new KeySelector[(String, Int), String] {
        override def getKey(value: (String, Int)): String = {
          value._1
        }
      })
      .sum(1)
      .print()

    env.execute()


  }

}
